package com.xaicode.template.rabbitmq.config;

import com.xaicode.template.rabbitmq.common.RabbitConst;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

/**
 * @author Locker
 * @date 2021/8/20 0020 16:50
 */
@Slf4j
@Configuration
public class RabbitConfig {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Bean
    public AmqpTemplate amqpTemplate() {
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setEncoding(StandardCharsets.UTF_8.toString());
        rabbitTemplate.setMandatory(true);

        /*
            ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调。
            即消息发送不到任何一个队列中  ack
         */
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) ->
                log.debug(String.format("- message [%s] send ok: reply code [%d], reply text [%s], exchange [%s], routing key [%s]",
                        message.getMessageProperties().getCorrelationId(), replyCode, replyText, exchange, routingKey)));

        /*
            ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。
            即消息发送到exchange  ack
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.debug("- send confirm to exchange with id [{}]", correlationData.getId());
            } else {
                log.debug("- send confirm to exchange fail, cause: {}", cause);
            }
        });
        return rabbitTemplate;
    }

    // ---- direct exchange ----

    @Bean("directExchange")
    public Exchange directExchange() {
        return ExchangeBuilder
                .directExchange(RabbitConst.DIRECT_EXCHANGE)
                // Set the durable flag
                .durable(true)
                .build();
    }

    @Bean("directQueue")
    public Queue directQueue() {
        return QueueBuilder.durable(RabbitConst.DIRECT_QUEUE).build();
    }

    /**
     * message from queue to exchange with routing key
     */
    @Bean
    public Binding directBinding(@Qualifier("directExchange") Exchange exchange,
                                 @Qualifier("directQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with(RabbitConst.DIRECT_ROUTING_KEY)
                .noargs();
    }

    // ---- fanout exchange ----

    @Bean("fanoutExchange")
    public FanoutExchange fanoutExchange() {
        return ExchangeBuilder
                .fanoutExchange(RabbitConst.FANOUT_EXCHANGE)
                .durable(true)
                .build();
    }

    @Bean("fanoutQueueA")
    public Queue fanoutQueueA() {
        return QueueBuilder.durable(RabbitConst.FANOUT_QUEUE_A).build();
    }

    @Bean("fanoutQueueB")
    public Queue fanoutQueueB() {
        return QueueBuilder.durable(RabbitConst.FANOUT_QUEUE_B).build();
    }

    @Bean("fanoutBindingA")
    public Binding fanoutBindingA(@Qualifier("fanoutQueueA") Queue queue,
                                  @Qualifier("fanoutExchange") FanoutExchange exchange) {
        return BindingBuilder
                .bind(queue)
                .to(exchange);
    }

    @Bean("fanoutBindingB")
    public Binding fanoutBindingB(@Qualifier("fanoutQueueB") Queue queue,
                                  @Qualifier("fanoutExchange") FanoutExchange exchange) {
        return BindingBuilder
                .bind(queue)
                .to(exchange);
    }

}
